﻿using Microsoft.Extensions.Logging;

using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace RabbitMQSample.WebApi.RabbitMQLibs
{
    public abstract class RabbitMQUnshareConsumerBase : BackgroundService, IDisposable
    {
        protected readonly IRabbitMQConnectionFactory _rabbitMQConnectionFactory;
        protected readonly ILogger _logger;
        private IConnection _connection;
        protected IModel _channel = null;
        protected CancellationToken cancellationToken;

        public RabbitMQUnshareConsumerBase(IRabbitMQConnectionFactory rabbitMQConnectionFactory, ILogger logger)
        {
            _rabbitMQConnectionFactory = rabbitMQConnectionFactory;
            _logger = logger;
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            cancellationToken = stoppingToken;

            try
            {
                while (!stoppingToken.IsCancellationRequested)
                {
                    try
                    {
                        await Task.Run(async () => await StartConsume(stoppingToken));
                    }
                    catch (RabbitMQ.Client.Exceptions.BrokerUnreachableException ex)
                    {
                        _logger.LogCritical(ex, "连接发生错误，稍后重试");
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, "消费发生错误，稍后重试");
                    }
                    await Task.Delay(5000, stoppingToken);
                }
            }
            catch (TaskCanceledException) { }
            catch (Exception ex)
            {
                _logger.LogError($"{nameof(ExecuteAsync)} 报错记录:{ex.StackTrace},{ex.Message}");
            }
        }

        protected virtual async Task StartConsume(CancellationToken stoppingToken)
        {
            #region 共享连接，不共享渠道
            if (_channel != null)
            {
                _channel.Dispose();
                _channel = null;
            }

            _channel = await _rabbitMQConnectionFactory.GetChannelAsync(stoppingToken);
            #endregion

            // 独立连接，独立渠道
            //_ = await GetChannelAsync(cancellationToken);

            using CancellationTokenSource cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
            _channel.ModelShutdown += (object sender, ShutdownEventArgs e) =>
            {
                cancellationTokenSource.Cancel();
                _logger.LogWarning($"Channel 断开，{e}");
            };

            _ = _channel.QueueDeclare(QueueName, true, false, false, default);
            if (!string.IsNullOrWhiteSpace(ExchangeName))
            {
                _channel.ExchangeDeclare(ExchangeName, ExchangeType, true, false, default);
                _channel.QueueBind(QueueName, ExchangeName, RoutingKey, default);
            }

            // prefetchSize：传递给消费者的消息的最大字节数；
            // prefetchCount：传递给消费者的消息的最大条数
            // global：是否对所有channel都有效
            _channel.BasicQos(0, PrefetchCount, false);
            AsyncEventingBasicConsumer consumer = new AsyncEventingBasicConsumer(_channel);

            consumer.Received += Consumer_Received;
            _channel.BasicConsume(QueueName, false, consumer);

            while (!cancellationTokenSource.Token.IsCancellationRequested && _channel?.IsOpen == true)
            {
                _logger.LogInformation($"保持连接，{DateTime.Now}");
                cancellationTokenSource.Token.WaitHandle.WaitOne(TimeSpan.FromMinutes(5));
            }

            consumer.Received -= Consumer_Received;

            if (_channel != null)
            {
                _channel.Dispose();
                _channel = null;
            }

            _logger.LogInformation("结束消息消费");
        }

        private async Task Consumer_Received(object sender, BasicDeliverEventArgs e)
        {
            if (cancellationToken == default || cancellationToken.IsCancellationRequested) return;

            AsyncEventingBasicConsumer _consumer = sender as AsyncEventingBasicConsumer;
            IModel channel = _consumer.Model;

            try
            {
                if (await HandlerMessage(e))
                {
                    if (_channel.IsOpen)
                        _channel.BasicAck(e.DeliveryTag, false);
                }
                else
                {
                    if (_channel.IsOpen)
                        _channel.BasicReject(e.DeliveryTag, true);
                }
            }
            catch (OperationCanceledException ex)
            {
                _logger.LogError(ex, ex.Message);
                if (_channel.IsOpen)
                    _channel.BasicReject(e.DeliveryTag, true);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, ex.Message);
                if (_channel.IsOpen)
                    _channel.BasicReject(e.DeliveryTag, true);
            }
        }

        /// <summary>
        /// 消息处理
        /// </summary>
        protected abstract Task<bool> HandlerMessage(BasicDeliverEventArgs e);

        /// <summary>
        /// 队列名称
        /// </summary>
        protected abstract string QueueName { get; }
        /// <summary>
        /// 交换机名称
        /// </summary>
        protected abstract string ExchangeName { get; }
        /// <summary>
        /// 交换机类型（RabbitMQ.Client.ExchangeType）
        /// </summary>
        protected abstract string ExchangeType { get; }
        /// <summary>
        /// 路由键
        /// </summary>
        protected abstract string RoutingKey { get; }
        /// <summary>
        /// 限流（表示一次性从队列中获取多少条消息，避免一次性发送的消息太多，导致消费端宕机）
        /// </summary>
        protected abstract ushort PrefetchCount { get; }

        public override void Dispose()
        {
            _channel?.Dispose();
            _connection?.Dispose();
            base.Dispose();
        }

        private async Task<IModel> GetChannelAsync(CancellationToken cancellationToken = default)
        {
            if (_channel != null && _channel.IsOpen) return _channel;
            _channel?.Dispose();

            if (_connection == null || _connection.IsOpen == false)
            {
                _connection?.Dispose();
            }

            _connection = await _rabbitMQConnectionFactory.GetConnectionAsync(cancellationToken);
            _channel = _connection.CreateModel();

            return _channel;
        }
    }
}
